package com.atguigu.gmall.realtime.app.dwd.db;

import com.atguigu.gmall.realtime.utils.HbaseUtil;
import com.atguigu.gmall.realtime.utils.MyKafkaUtil;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

/**
 * @author Felix
 * @date 2023/10/7
 * 评论事实表
 * 需要启动的进程
 *      zk、kafka、maxwell、hdfs、hbase、DwdInteractionCommentInfo
 */
public class DwdInteractionCommentInfo {
    public static void main(String[] args) {
        //TODO 1.基本环境准备
        //1.1 指定流处理环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        //1.2 设置并行度
        env.setParallelism(4);
        //1.3 指定表执行环境
        StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);
        /*//TODO 2.检查点相关的设置
        //2.1 开启检查点
        env.enableCheckpointing(5000L, CheckpointingMode.EXACTLY_ONCE);
        //2.2 设置检查点超时时间
        env.getCheckpointConfig().setCheckpointTimeout(60000L);
        //2.3 设置job取消后检查点是否保留
        env.getCheckpointConfig().setExternalizedCheckpointCleanup(CheckpointConfig.ExternalizedCheckpointCleanup.RETAIN_ON_CANCELLATION);
        //2.4 设置两个检查点之间最小时间间隔
        env.getCheckpointConfig().setMinPauseBetweenCheckpoints(2000L);
        //2.5 设置重启策略
        env.setRestartStrategy(RestartStrategies.failureRateRestart(3, Time.days(30),Time.seconds(3)));
        //2.6 设置状态后端
        env.setStateBackend(new HashMapStateBackend());
        env.getCheckpointConfig().setCheckpointStorage("hdfs://hadoop102:8020/gmall/ck");
        //2.7 设置操作hadoop的用户
        System.setProperty("HADOOP_USER_NAME","atguigu");*/
        //TODO 3.从kafka的主题中读取数据 创建动态表----kafka连接器
        //3.1 声明消费者组
        String groupId = "dwd_interaction_comment_group";
        //3.2 执行建表
        tableEnv.executeSql(MyKafkaUtil.getTopicDbDDL(groupId));
        //tableEnv.executeSql("select * from topic_db").print();

        //TODO 4.将评论数据过滤出来  --- table='comment_info' and type='insert'
        Table commentTable = tableEnv.sqlQuery("select \n" +
            "    `data`['id'] id,\n" +
            "    `data`['user_id'] user_id,\n" +
            "    `data`['sku_id'] sku_id,\n" +
            "    `data`['appraise'] appraise,\n" +
            "    `data`['comment_txt'] comment_txt,\n" +
            "    ts,\n" +
            "    proc_time\n" +
            "from topic_db where `table`='comment_info' and `type`='insert'");
        tableEnv.createTemporaryView("comment_table",commentTable);
        // tableEnv.executeSql("select * from comment_table").print();
        //TODO 5.从hbase中读取字典表数据 创建动态表 --- hbase连接器
        tableEnv.executeSql(HbaseUtil.getBaseDicDDL());

        // tableEnv.executeSql("select dic_code,dic_name from base_dic").print();
        //TODO 6.关联评论表和字典表  --- lookupJoin
        Table joinedTable = tableEnv.sqlQuery("SELECT\n" +
            " id,\n" +
            " user_id,\n" +
            " sku_id,\n" +
            " appraise,\n" +
            " dic.dic_name appraise_name,\n" +
            " comment_txt,\n" +
            " ts\n" +
            "FROM comment_table AS c\n" +
            "  JOIN base_dic FOR SYSTEM_TIME AS OF c.proc_time AS dic\n" +
            "    ON c.appraise = dic.dic_code");
        tableEnv.createTemporaryView("joined_table",joinedTable);
        // tableEnv.executeSql("select * from joined_table").print();
        //TODO 7.将关联的结果写到kafka主题 --- upsert-kafka
        //7.1 创建动态表和要写入的主题进行关联
        tableEnv.executeSql("CREATE TABLE dwd_interaction_comment (\n" +
            "     id string,\n" +
            "     user_id string,\n" +
            "     sku_id string,\n" +
            "     appraise string,\n" +
            "     appraise_name string,\n" +
            "     comment_txt string,\n" +
            "     ts string,\n" +
            "  PRIMARY KEY (id) NOT ENFORCED\n" +
            ") " + MyKafkaUtil.getUpsertKafkaDDL("dwd_interaction_comment"));
        //7.2 写入
        tableEnv.executeSql("insert into dwd_interaction_comment select * from joined_table");
    }
}
